package cn.itcast.flink.window.time;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
/**
* 窗口统计案例演示：滚动计数窗口（Tumbling Count Window)，数字累加求和统计
*/
public class StreamTumblingCountWindow_2 {
    public static void main(String[] args) throws Exception {
        // 1. 执行环境-env：流计算执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1) ;
        // 2. 数据源-source：Socket接收数据
        DataStreamSource<String> inputStream = env.socketTextStream("node1.itcast.cn", 9999);
        // 3. 转换处理-transformation：调用DataStream函数，处理数据
        SingleOutputStreamOperator<Integer> numStream = inputStream
                .filter(line -> null != line && line.trim().length() > 0)
                //转换整数
                .map(line -> Integer.parseInt(line.trim()));
        //TODO:每隔5秒统计一次,并进行累加之和
        SingleOutputStreamOperator<Integer> sumStream = numStream
                .countWindowAll(5)
                .sum(0);

        // 4. 数据终端-sink
        sumStream.printToErr("sum");
        // 5. 触发执行-execute
        env.execute(StreamTumblingCountWindow_2.class.getSimpleName()) ;
} }